package com.stars.easyms.mq.core;

import com.alibaba.fastjson.JSON;
import com.stars.easyms.base.bean.LazyLoadBean;
import com.stars.easyms.mq.enums.RedeliveryPolicy;
import com.stars.easyms.mq.asyn.AsynSendMQMessage;
import com.stars.easyms.mq.constant.MQConstants;
import com.stars.easyms.mq.dto.MQMessageDTO;
import com.stars.easyms.mq.enums.DestinationType;
import com.stars.easyms.mq.enums.MQType;
import com.stars.easyms.mq.properties.EasyMsMQProperties;
import com.stars.easyms.mq.service.MQMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;

/**
 * MQ消息发送类
 *
 * @author guoguifang
 * @date 2018-04-23 13:54
 * @since 1.0.0
 */
public class EasyMsMQSender {

    private static final Logger logger = LoggerFactory.getLogger(EasyMsMQSender.class);

    /**
     * 懒加载的方式，防止没有发送MQ消息的时候也创建异步线程池
     */
    private final LazyLoadBean<AsynSendMQMessage> asyncSendMQMessage = new LazyLoadBean<>(AsynSendMQMessage::new);

    private final MQMessageService mqMessageService;

    private final String applicationId;

    /**
     * 发送MQ消息默认方法：点对点消息、重试策略为指数策略、重试间隔为60秒
     *
     * @param key 发送目标:ActiveMQ对应destination，RocketMQ对应topic
     * @param obj 消息对象
     */
    public void send(String key, Object obj) {
        send(key, DestinationType.TOPIC, RedeliveryPolicy.EXPONENTIAL, 60, obj);
    }

    /**
     * 发送MQ消息方法
     *
     * @param key             发送目标:ActiveMQ对应destination，RocketMQ对应topic
     * @param destinationType 发送类型：点对点、广播
     * @param obj             消息对象
     */
    public void send(String key, DestinationType destinationType, Object obj) {
        send(key, destinationType, RedeliveryPolicy.EXPONENTIAL, 60, obj);
    }

    /**
     * 发送MQ消息方法：点对点个
     *
     * @param key              发送目标:ActiveMQ对应destination，RocketMQ对应topic
     * @param redeliveryPolicy 重试策略
     * @param redeliveryDelay  重试间隔
     * @param obj              消息对象
     */
    public void send(String key, RedeliveryPolicy redeliveryPolicy, long redeliveryDelay, Object obj) {
        send(key, DestinationType.TOPIC, redeliveryPolicy, redeliveryDelay, obj);
    }

    /**
     * 发送MQ消息方法
     *
     * @param key              发送目标:ActiveMQ对应destination，RocketMQ对应topic
     * @param destinationType  发送类型：点对点、广播
     * @param redeliveryPolicy 重试策略
     * @param redeliveryDelay  重试间隔
     * @param obj              消息对象
     */
    public void send(String key, DestinationType destinationType, RedeliveryPolicy redeliveryPolicy, long redeliveryDelay, Object obj) {
        // 将参数转换成字符串
        String message;
        if (obj == null) {
            return;
        } else if (obj instanceof String) {
            message = (String) obj;
        } else {
            message = JSON.toJSONString(obj);
        }

        // 记录待发送消息日志
        logger.info("待发送MQ消息到目标({}),消息内容:{}", key, message);

        // 采用异步处理机制处理MQ消息发送
        MQMessageDTO mqMessageDTO = new MQMessageDTO();
        mqMessageDTO.setId(mqMessageService.getId());
        mqMessageDTO.setMessageId(mqMessageService.getMessageId());
        mqMessageDTO.setType(MQType.SEND.getCode());
        mqMessageDTO.setDestinationType(destinationType.getCode());
        mqMessageDTO.setKey(key);
        mqMessageDTO.setSourceSys(applicationId);
        mqMessageDTO.setMessage(message);
        mqMessageDTO.setStatus(MQConstants.MQ_STATUS_PROCESSING);
        mqMessageDTO.setFailCount(0L);
        mqMessageDTO.setRedeliveryPolicy(redeliveryPolicy.getCode());
        mqMessageDTO.setRedeliveryDelay(redeliveryDelay <= 0 ? 0 : redeliveryDelay);
        mqMessageDTO.setFireTime(new Timestamp(System.currentTimeMillis()));
        asyncSendMQMessage.accept(a -> a.add(mqMessageDTO));
    }

    public EasyMsMQSender(MQMessageService mqMessageService, String applicationId, EasyMsMQProperties easyMsMQProperties) {
        this.mqMessageService = mqMessageService;
        this.applicationId = applicationId;
        if (easyMsMQProperties.isEnabled()) {
            asyncSendMQMessage.getBean();
        }
    }

}
